package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.UserActionBean;
import com.atguigu.util.ClickHouseUtil;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

//用户行为漏斗分析
public class DwsUserAction {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        // TODO 3 从kafka的读取dwd_traffic_home_and_good_detail主题数据
        String topicName = "dwd_traffic_home_and_good_detail";
        String groupId = "dws_user_action";
        DataStreamSource<String> homeAndGoodDetailStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));
        String topicName1 = "dwd_trade_cart_add";
        DataStreamSource<String> cartAddStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName1, groupId));
        String topicName2 = "dwd_course_order_detail";
        DataStreamSource<String> orderDetailStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName2, groupId));
        String topicName3 = "dwd_trade_pay_detail_suc";
        DataStreamSource<String> payDetailSucStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName3, groupId));

        // TODO 4 转换格式
        SingleOutputStreamOperator<UserActionBean> beanStream = homeAndGoodDetailStream.flatMap(new FlatMapFunction<String, UserActionBean>() {
            @Override
            public void flatMap(String s, Collector<UserActionBean> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                Long ts = jsonObject.getLong("ts");
                long homeUserCt = 0L;
                long goodDetailUserCt = 0L;
                if ("home".equals(pageId)) {
                    homeUserCt = 1L;
                    collector.collect(new UserActionBean("", "", homeUserCt, goodDetailUserCt, 0L, 0L, 0L, ts));
                } else {
                    goodDetailUserCt = 1L;
                    collector.collect(new UserActionBean("", "", homeUserCt, goodDetailUserCt, 0L, 0L, 0L, ts));
                }
            }
        });
        SingleOutputStreamOperator<UserActionBean> beanCartStream = cartAddStream.flatMap(new FlatMapFunction<String, UserActionBean>() {
            @Override
            public void flatMap(String s, Collector<UserActionBean> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Long ts = jsonObject.getLong("ts");
                collector.collect(new UserActionBean("", "", 0L, 0L, 1L, 0L, 0L, ts));
            }
        });
        SingleOutputStreamOperator<UserActionBean> beanOrderStream = orderDetailStream.flatMap(new FlatMapFunction<String, UserActionBean>() {
            @Override
            public void flatMap(String s, Collector<UserActionBean> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Long ts = jsonObject.getLong("ts");
                collector.collect(new UserActionBean("", "", 0L, 0L, 0L, 1L, 0L, ts));
            }
        });
        SingleOutputStreamOperator<UserActionBean> beanPaySucStream = payDetailSucStream.flatMap(new FlatMapFunction<String, UserActionBean>() {
            @Override
            public void flatMap(String s, Collector<UserActionBean> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Long ts = jsonObject.getLong("ts");
                collector.collect(new UserActionBean("", "", 0L, 0L, 0L, 0L, 1L, ts));
            }
        });

        // TODO 5 Union
        DataStream<UserActionBean> unionStream = beanStream.union(beanCartStream).union(beanOrderStream).union(beanPaySucStream);

        // TODO 6 开窗聚合
        SingleOutputStreamOperator<Object> reduceStream = unionStream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserActionBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserActionBean>() {
                    @Override
                    public long extractTimestamp(UserActionBean userActionBean, long l) {
                        return userActionBean.getTs() * 1000L;
                    }
                })).windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<UserActionBean>() {
                    @Override
                    public UserActionBean reduce(UserActionBean v1, UserActionBean v2) throws Exception {
                        v1.setHomeUserCt(v1.getHomeUserCt() + v2.getHomeUserCt());
                        v1.setGoodDetailUserCt(v1.getGoodDetailUserCt() + v2.getGoodDetailUserCt());
                        v1.setCartAddUserCt(v1.getCartAddUserCt() + v2.getCartAddUserCt());
                        v1.setOrderUserCt(v1.getOrderUserCt() + v2.getOrderUserCt());
                        v1.setPaymentSucUserCt(v1.getPaymentSucUserCt() + v2.getPaymentSucUserCt());
                        return v1;
                    }
                }, new AllWindowFunction<UserActionBean, Object, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<UserActionBean> iterable, Collector<Object> collector) throws Exception {
                        UserActionBean next = iterable.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        collector.collect(next);
                    }
                });

        // TODO 7 写出到clickhouse中
        reduceStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_user_action_window values (?,?,?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupId);
    }
}
